package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

public class Code05ScanTable {

    Connection connection;
    @Before
    public void getConnection(){
        Configuration configuration = new Configuration();

        // 注意：需要在当前平台中配置node1 node2的主机映射
        configuration.set("hbase.zookeeper.quorum","192.168.163.100,node1,node2");
        try {
            connection  = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    @Test
    public void scanTable() throws IOException {

        Table table = connection.getTable(TableName.valueOf("api:stu"));

        // scan '表名'
        Scan scan = new Scan();

        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            // result就是一个RowKey对应的结果数据
            printRes(result);
        }
    }
    @Test
    public void scanTable2() throws IOException {

        Table table = connection.getTable(TableName.valueOf("api:stu"));

        // scan '表名'
        Scan scan = new Scan();
//        scan.setStartRow()
//        scan.withStartRow("1500100001".getBytes());
//        scan.withStopRow("1500101000~".getBytes());


//        scan.withStartRow("1500101000~".getBytes());
        scan.withStopRow("1500100001".getBytes());

        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            // result就是一个RowKey对应的结果数据
            printRes(result);
        }
    }



    public  void printRes(Result result){
        List<Cell> cells = result.listCells();
        String rowKey = Bytes.toString(result.getRow());
        System.out.print(rowKey+"\t");
        for (Cell cell : cells) {
//                cell.getValueArray()
            // 通过CellUtil工具类对cell对象中的数据进行取值
            String family = Bytes.toString(CellUtil.cloneFamily(cell));
            String colum = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
//                System.out.println("family:"+family+"\tcol:"+colum+"\tvalue:"+value);
            // 对于添加的数据是按照 rowKey + 列名称 + 时间戳 进行排序的
            System.out.print(value+"\t");
        }
        System.out.println();
    }


    @After
    public void close(){
        try {
            connection.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
